package com.pancm.datacollector.processor;

import com.pancm.datacollector.forwarder.DataForwarder;
import com.pancm.datacollector.queue.CollectedData;
import com.pancm.datacollector.queue.DataQueueManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 数据处理管理器
 * 负责从队列获取数据并交给处理器处理
 */
@Component
public class DataProcessorManager {
    private static final Logger LOG = LoggerFactory.getLogger(DataProcessorManager.class);

    private final DataQueueManager queueManager;
    private final DataProcessor processor;
    private final AtomicBoolean running;
    private ExecutorService executorService;

    @Autowired
    private ProcessorProperties properties;

    @Autowired
    private DataForwarder forwarder;

    public DataProcessorManager(DataQueueManager queueManager, DataProcessor processor) {
        this.queueManager = queueManager;
        this.processor = processor;
        this.running = new AtomicBoolean(false);
    }

    @PostConstruct
    public void start() {
        this.executorService = Executors.newFixedThreadPool(properties.getThread().getCount());
        
        LOG.info("启动数据处理管理器，处理线程数: {}", properties.getThread().getCount());
        running.set(true);
        
        // 启动多个处理线程
        for (int i = 0; i < properties.getThread().getCount(); i++) {
            executorService.submit(this::processLoop);
        }
    }

    @PreDestroy
    public void stop() {
        LOG.info("停止数据处理管理器...");
        running.set(false);
        
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(
                properties.getThread().getShutdown().getTimeout(), 
                TimeUnit.MILLISECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            LOG.error("等待处理器关闭被中断", e);
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 处理循环
     * 不断从队列获取数据并处理
     */
    private void processLoop() {
        Thread.currentThread().setName("DataProcessor-" + Thread.currentThread().getId());
        LOG.info("处理线程启动: {}", Thread.currentThread().getName());

        while (running.get()) {
            try {
                // 从队列获取数据
                CollectedData data = queueManager.poll();
                if (data == null) {
                    continue;
                }

                // 处理数据
                try {
                    processor.process(data);
                    LOG.debug("数据处理完成, ID: {}", data.getId());
                    
                    // 数据处理完成后，进行转发
                    try {
                        forwarder.forward(data.getData());
                        LOG.debug("数据转发完成, ID: {}", data.getId());
                    } catch (Exception e) {
                        LOG.error("数据转发失败, ID: {}", data.getId(), e);
                    }
                    
                } catch (Exception e) {
                    LOG.error("数据处理失败, ID: {}", data.getId(), e);
                }
            } catch (Exception e) {
                LOG.error("处理循环发生异常", e);
            }
        }

        LOG.info("处理线程退出: {}", Thread.currentThread().getName());
    }

    /**
     * 获取当前是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }
} 